
#include "RtBase.h"
#include "RtTransportTcp.h"
#include "RtInetAddr.h"
#include "RtMessageBlock.h"
#include "RtReactorInterface.h"

CRtTransportTcp::CRtTransportTcp(IRtReactor *pReactor)
	: CRtTransportBase(pReactor)
	, m_needOnSend(FALSE)
{
}

CRtTransportTcp::~CRtTransportTcp()
{
	Close_t(RT_OK);
}

RtResult CRtTransportTcp::Open_t()
{
	RtResult rv = RT_ERROR_NETWORK_SOCKET_ERROR;
	
	DWORD dwRcv = 65535, dwSnd = 65535;
	rv = SetOption(RT_OPT_TRANSPORT_RCV_BUF_LEN, &dwRcv);
//	RT_ASSERTE(RT_SUCCEEDED(rv));
	rv = SetOption(RT_OPT_TRANSPORT_SND_BUF_LEN, &dwSnd);
//	RT_ASSERTE(RT_SUCCEEDED(rv));

	int nNoDelay = 1;
	if (m_socketTcp.SetOption(IPPROTO_TCP, TCP_NODELAY, &nNoDelay, sizeof(nNoDelay)) == -1) 
	{
		RT_ERROR_TRACE("CRtTransportTcp::Open_t, SetOption(TCP_NODELAY) failed! err=" << errno);
	}

	rv = m_pReactor->RegisterHandler(this, IRtEventHandler::READ_MASK | IRtEventHandler::WRITE_MASK);
	if (RT_FAILED(rv) && rv != RT_ERROR_FOUND) 
	{
		RT_ERROR_TRACE_THIS("CRtTransportTcp::Open_t, RegisterHandler(READ_MASK|WRITE_MASK) failed!");
		return rv;
	}
	else
		return RT_OK;
}

RT_HANDLE CRtTransportTcp::GetHandle() const
{
	return m_socketTcp.GetHandle();
}

int CRtTransportTcp::OnInput(RT_HANDLE )
{
	static char szBuf[65536*2];
#if 1
	int nRecv = Recv_i(szBuf, sizeof(szBuf));
//	RT_INFO_TRACE_THIS("CRtTransportTcp::OnInput, recv="<<nRecv<<" fd=" << m_socketTcp.GetHandle()<<" err="<<errno);
	if (nRecv <= 0)
		return nRecv;

/***********Jobs Add at 2008.06.03 for flow control************************/
// 	m_dwTotalRecv += nRecv;
/***********Jobs Add at 2008.06.03 end************************/

	CRtMessageBlock mbOn(
		nRecv, 
		szBuf, 
		CRtMessageBlock::DONT_DELETE | CRtMessageBlock::WRITE_LOCKED, 
		nRecv);
	
	if (m_pSink)
		m_pSink->OnReceive(mbOn, this);

	return 0;
#else
	while (1) {
		int nRecv = Recv_i(szBuf, sizeof(szBuf));
//		RT_INFO_TRACE_THIS("CRtTransportTcp::OnInput, recv="<<nRecv<<" fd=" << m_socketTcp.GetHandle()<<" err="<<errno);
		if (nRecv <= 0)
			return nRecv;

/***********Jobs Add at 2008.06.03 for flow control************************/
// 		m_dwTotalRecv += nRecv;
/***********Jobs Add at 2008.06.03 end************************/

		CRtMessageBlock mbOn(
			nRecv, 
			szBuf, 
			CRtMessageBlock::DONT_DELETE | CRtMessageBlock::WRITE_LOCKED, 
			nRecv);
	
		if (m_pSink)
			m_pSink->OnReceive(mbOn, this);
	}
	return 0;
#endif
}

int CRtTransportTcp::OnOutput(RT_HANDLE aFd)
{
	if (!m_needOnSend || m_socketTcp.GetHandle()==RT_INVALID_HANDLE) 
		return 0;

	//RT_INFO_TRACE_THIS("CRtTransportTcp::OnOutput, fd="<<aFd);

	m_needOnSend = FALSE;
	if (m_pSink) 
		m_pSink->OnSend(this);
	return 0;
}

CRtSocketStream& CRtTransportTcp::GetPeer()
{
	return m_socketTcp;
}

RtResult CRtTransportTcp::Close_t(RtResult aReason)
{
	if (m_socketTcp.GetHandle() != RT_INVALID_HANDLE) {
		m_pReactor->RemoveHandler(this);
		m_socketTcp.Close(aReason);
	}
	return RT_OK;
}

RtResult CRtTransportTcp::SendData(
		CRtMessageBlock& inData,
		BOOL inSendImmediately,
		BOOL inDestroyData,
		DataPriority inPriority, 
		BOOL inNeedAck)
{
	if (m_socketTcp.GetHandle() == RT_INVALID_HANDLE) {
//		RT_WARNING_TRACE_THIS("CRtTransportTcp::SendData, socket is invalid.");
		return RT_ERROR_NOT_AVAILABLE;
	}

	if (m_needOnSend)
		return RT_ERROR_PARTIAL_DATA;

	const CRtMessageBlock* tmpData = &inData;
	static iovec iov[RT_IOV_MAX];
	DWORD iovNum = 0;
	DWORD fillLen = 0;
	int sentTotal = 0;
	int rv = 0;

	do
	{
		iovNum = tmpData->FillIov(iov, RT_IOV_MAX, fillLen, tmpData);
		if (iovNum == 0)
			break;

		rv = m_socketTcp.SendV(iov, iovNum);
		if (rv < 0) {
			if (errno == EWOULDBLOCK) {
				m_needOnSend = TRUE;
				break;
			}
			else {
				RT_WARNING_TRACE_THIS("CRtTransportTcp::SendData, sendv() failed! err=" << RtGetSystemErrorInfo(errno) );
				return RT_ERROR_NETWORK_SOCKET_ERROR;
			}
		}
		sentTotal += rv;
		if ((DWORD)rv < fillLen)
		{
			m_needOnSend = TRUE;
			break;
		}
	} while (tmpData);

	if (m_needOnSend)
	{
		inData.AdvanceChainedReadPtr(sentTotal);
		return RT_ERROR_PARTIAL_DATA;
	}

	if (inDestroyData)
		inData.DestroyChained();
	
	return RT_OK;
}

RtResult CRtTransportTcp::GetOption(DWORD aCommand, LPVOID aArg)
{
	switch (aCommand) {
	case RT_OPT_TRANSPORT_FIO_NREAD:
		if (m_socketTcp.Control(FIONREAD, aArg) == -1) {
			RT_WARNING_TRACE_THIS("CRtTransportTcp::GetOption, (RT_OPT_TRANSPORT_FIO_NREAD) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		return RT_OK;

	case RT_OPT_TRANSPORT_FD:
		*(static_cast<RT_HANDLE *>(aArg)) = m_socketTcp.GetHandle();
		return RT_OK;

	case RT_OPT_TRANSPORT_LOCAL_ADDR:
		if (m_socketTcp.GetLocalAddr(*(static_cast<CRtInetAddr*>(aArg))) == -1) {
			RT_WARNING_TRACE_THIS("CRtTransportTcp::GetOption, (RT_OPT_TRANSPORT_LOCAL_ADDR) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;

	case RT_OPT_TRANSPORT_PEER_ADDR:
		if (m_peerAddrInProxyCase.GetPort()!=0)
		{
			// In the case of connection via proxy. 
			*(static_cast<CRtInetAddr*>(aArg)) = m_peerAddrInProxyCase;
			return RT_OK;
		}
		if (m_socketTcp.GetRemoteAddr(*(static_cast<CRtInetAddr*>(aArg))) == -1) {
			RT_WARNING_TRACE_THIS("CRtTransportTcp::GetOption, (RT_OPT_TRANSPORT_PEER_ADDR) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;

	case RT_OPT_TRANSPORT_SOCK_ALIVE: {
		if (m_socketTcp.GetHandle() == RT_INVALID_HANDLE) {
			*static_cast<BOOL*>(aArg) = FALSE;
			return RT_ERROR_NOT_INITIALIZED;
		}
		char cTmp;
		int nRet = m_socketTcp.Recv(&cTmp, sizeof(cTmp), MSG_PEEK);
		if (nRet > 0 || (nRet < 0 && errno == EWOULDBLOCK))
			*static_cast<BOOL*>(aArg) = TRUE;
		else
			*static_cast<BOOL*>(aArg) = FALSE;
		return RT_OK;
	}
		
	case RT_OPT_TRANSPORT_TRAN_TYPE:
		*(static_cast<CRtConnectionManager::CType*>(aArg)) 
			= CRtConnectionManager::CTYPE_TCP;
		return RT_OK;

	case RT_OPT_TRANSPORT_SND_BUF_LEN: {
		int nLen = sizeof(DWORD);
		if (m_socketTcp.GetOption(SOL_SOCKET, SO_SNDBUF, aArg, &nLen) == -1) {
//			RT_ERROR_TRACE_THIS("CRtTransportTcp::GetOption, GetOption(SO_SNDBUF) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;
									   }

	case RT_OPT_TRANSPORT_RCV_BUF_LEN: {
		int nLen = sizeof(DWORD);
		if (m_socketTcp.GetOption(SOL_SOCKET, SO_RCVBUF, aArg, &nLen) == -1) {
//			RT_ERROR_TRACE_THIS("CRtTransportTcp::GetOption, GetOption(SO_RCVBUF) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;
									   }

	default:
		return CRtTransportBase::GetOption(aCommand, aArg);
	}
}

RtResult CRtTransportTcp::SetOption(DWORD aCommand, LPVOID aArg)
{
	RT_ASSERTE_RETURN(aArg, RT_ERROR_INVALID_ARG);
	switch (aCommand) {
	case RT_OPT_TRANSPORT_FD: {
		// we allow user to set TCP socket to RT_INVALID_HANDLE, 
		// mainly used by CRtConnectorProxyT.
		RT_HANDLE hdNew = *(static_cast<RT_HANDLE *>(aArg));
		RT_ASSERTE_RETURN(hdNew == RT_INVALID_HANDLE, RT_ERROR_INVALID_ARG);
		m_socketTcp.SetHandle(hdNew);
		return RT_OK;
							  }

	case RT_OPT_TRANSPORT_SND_BUF_LEN:
		if (m_socketTcp.SetOption(SOL_SOCKET, SO_SNDBUF, aArg, sizeof(DWORD)) == -1) {
//			RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(SO_SNDBUF) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;

	case RT_OPT_TRANSPORT_RCV_BUF_LEN:
		if (m_socketTcp.SetOption(SOL_SOCKET, SO_RCVBUF, aArg, sizeof(DWORD)) == -1) {
//			RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(SO_RCVBUF) failed! err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
		else
			return RT_OK;

	case RT_OPT_TRANSPORT_TCP_KEEPALIVE: {
		DWORD dwTime = *static_cast<DWORD*>(aArg);
		int nKeep = dwTime > 0 ? 1 : 0;
		if (m_socketTcp.SetOption(SOL_SOCKET, SO_KEEPALIVE, &nKeep, sizeof(nKeep)) == -1) {
			RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(SO_KEEPALIVE) failed!"
				" dwTime=" << dwTime << 
				" err=" << errno);
			return RT_ERROR_NETWORK_SOCKET_ERROR;
		}
#ifdef RT_LINUX
		if (dwTime > 0) {
			if (m_socketTcp.SetOption(SOL_TCP, TCP_KEEPIDLE, &dwTime, sizeof(dwTime)) == -1) {
				RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(TCP_KEEPINTVL) failed!"
					" dwTime=" << dwTime << 
					" err=" << errno);
				return RT_ERROR_NETWORK_SOCKET_ERROR;
			}
		}
#endif // RT_LINUX
		return RT_OK;
	}
		
	case RT_OPT_TRANSPORT_TCP_KEEPINTVL:{		
		DWORD dwTime = *static_cast<DWORD*>(aArg);
		
#ifdef RT_LINUX
		if (dwTime > 0) {
			if (m_socketTcp.SetOption(SOL_TCP, TCP_KEEPINTVL, &dwTime, sizeof(dwTime)) == -1) {
				RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(TCP_KEEPINTVL) failed!"
					" dwTime=" << dwTime << 
					" err=" << errno);
				return RT_ERROR_NETWORK_SOCKET_ERROR;
			}
		}
#endif // RT_LINUX
		return RT_OK;
	}
	case RT_OPT_TRANSPORT_TCP_KEEPCNT:{		
		DWORD dwTime = *static_cast<DWORD*>(aArg);
		
#ifdef RT_LINUX
		if (dwTime > 0) {
			if (m_socketTcp.SetOption(SOL_TCP, TCP_KEEPCNT, &dwTime, sizeof(dwTime)) == -1) {
				RT_ERROR_TRACE_THIS("CRtTransportTcp::SetOption, SetOption(TCP_KEEPCNT) failed!"
					" dwTime=" << dwTime << 
					" err=" << errno);
				return RT_ERROR_NETWORK_SOCKET_ERROR;
			}
		}
#endif // RT_LINUX
		return RT_OK;
	}
	case RT_OPT_TRANSPORT_TOS: 
		return SetTos2Socket(m_socketTcp, aArg);

	case RT_OPT_TRANSPORT_PEER_ADDR:
		// In the case of connection via proxy. 
		m_peerAddrInProxyCase = *(static_cast<CRtInetAddr*>(aArg));
		return RT_OK;

	default:
		RT_WARNING_TRACE_THIS("CRtTransportTcp::SetOption,"
			" unknow aCommand=" << aCommand << 
			" aArg=" << aArg);
		return RT_ERROR_INVALID_ARG;
	}
}
